package mspbots.cw.client.core.data.reports;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.extern.log4j.Log4j2;
import mspbots.cw.client.core.AbstractToolsUtil;
import mspbots.cw.client.core.client.ClientProperties;
import mspbots.cw.client.core.data.BaseSyncService;
import mspbots.cw.common.*;
import mspbots.cw.common.report.ReportIdIndex;
import mspbots.cw.common.report.ReportNames;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * mspbots.cw.client.core.data.reports.ReportsServiceImpl
 *
 * @author <a href="https://github.com/vnobo">Alex bob</a>
 * @date Created by 2020/11/10
 */
@Log4j2
@Service
public class ReportsServiceImpl extends AbstractToolsUtil implements BaseSyncService {


    private final ReportNames reportNames = ReportNames.names();
    private final ReportIdIndex idIndex = ReportIdIndex.ids();

  /*  @Override
    public Mono<Void> syncFast(ClientProperties properties) {
        return Flux.fromStream(reportNames.keySet().parallelStream()
                .filter(k -> "TicketAudit".equals(k) || "TimeAudit".equals(k)))
                .delayElements(Duration.ofSeconds(2))
                .flatMap(key -> this.filterCacheGet(properties.getTenantId(),
                        key, reportNames.get(key)))
                .contextWrite(ctx -> ctx.put("properties", properties))
                .then();
    }*/


    @Override
    public Mono<Void> syncSlower(ClientProperties properties) {
        return Flux.fromStream(reportNames.keySet().parallelStream())
                .filter("Service_SLA"::equals)
                .delayElements(Duration.ofSeconds(2))
                .flatMap(key -> this.filterCacheGet(properties.getTenantId(),
                        key, reportNames.get(key)))
                .contextWrite(ctx -> ctx.put("properties", properties))
                .then();
    }

    private Flux<JsonNode> filterCacheGet(Long tenantId, String reportName, String filterName) {
        ResultSync resultSync = ResultSync.withDefault(reportName + "-REPORTS-SYNC");
        String redisCacheKey = RedisKeys.REPORT_ANCHOR_REDIS_KEY_PREFIX + tenantId + ":" + reportName;
        return this.stringRedisTemplate.opsForValue().get(redisCacheKey)
                .map(str -> LocalDateTime.parse(str, DateTimeFormatter.ISO_DATE_TIME))
                .defaultIfEmpty(LocalDateTime.now().minusDays(2))
                .flatMapMany(syncTime -> this.fetch(reportName, QueryParams.withDefault()
                        .setConditions(filterName + ">[" + syncTime.format(DateTimeFormatter.ISO_DATE_TIME) + "]")
                        .setOrderBy(filterName + " asc")))
                .contextWrite(ctx -> ctx.put("resultSync", resultSync))
                .doOnNext(resultNode -> this.syncResult(tenantId,
                        resultSync.getRequestId(), reportName, resultNode))
                .doOnNext(resultNode -> this.recordAnchorTime(redisCacheKey,
                        resultNode.findValue(filterName).textValue()));
    }

    private void syncResult(Long tenantId, UUID uuid, String name, JsonNode resultNode) {
        this.sendEntity(RabbitKeys.REPORTS_SYNC_QUEUE_KEY, ResultMsg.builder()
                .requestId(uuid).id(resultNode.get("id").longValue())
                .action(name).entity(resultNode).tenantId(tenantId)
                .syncDate(LocalDateTime.now()).build());
    }

    private Flux<JsonNode> fetch(String reportName, QueryParams queryParams) {
        log.debug("Get reports name: [{}], params: [{}].", reportName, queryParams);
        return super.cwWebClient.fetch("/system/reports/{reportName}", queryParams,
                Map.of("reportName", reportName))
                .flatMap(res -> this.buildReports(reportName, res));
    }

    private Flux<JsonNode> buildReports(String reportName, JsonNode resultNode) {
        JsonNode columnNode = resultNode.withArray("column_definitions");
        List<String> header = new ArrayList<>();
        columnNode.forEach(node -> node.fieldNames().forEachRemaining(header::add));
        JsonNode rowNode = resultNode.withArray("row_values");
        return Flux.fromIterable(rowNode).map(jsonNode -> {
            ObjectNode reportNode = this.objectMapper.createObjectNode();
            for (int x = 0; x < jsonNode.size(); x++) {
                reportNode.set(header.get(x), jsonNode.get(x));
            }
            return idIndex.keySet().parallelStream().filter(k -> k.equals(reportName))
                    .map(idIndex::get)
                    .map(id -> {
                        reportNode.set("id", jsonNode.get(id));
                        return reportNode;
                    }).findAny().orElse(null);
        });
    }

    private Flux<String> getName(QueryParams queryParams) {
        log.debug("Get reports name params: [{}].", queryParams);
        return super.cwWebClient.get("/system/reports", queryParams)
                .flatMapMany(Flux::fromIterable)
                .map(jsonNode -> jsonNode.get("name").textValue());
    }
}
